package com.example.mqtt_mode.mqtt.config;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;

/**
 * @author IT空门_门主
 * @date 2024/1/4
 * 配置连接
 */

@Slf4j
@PropertySource("classpath:application.yml")
@Component
public class MqttConfig  {

    private static String broker;
    @Value("${mqtt.broker}")
    public void setBroker(String broker) {
        this.broker = broker;
    }

    private static String topic;
    @Value("${mqtt.topic}")
    public void setTopic(String topic) {
        this.topic = topic;
    }


    private static String username;
    @Value("${mqtt.username}")
    public void setUsername(String username) {
        this.username = username;
    }

    private static String password;
    @Value("${mqtt.password}")
    public void setPassword(String password) {
        this.password = password;
    }


    private static String clientid;
    @Value("${mqtt.clientid}")
    public void setClientid(String clientid) {
        this.clientid = clientid;
    }


    private static Integer qos;
    @Value("${mqtt.qos}")
    public void setQos(Integer qos) {
        this.qos = qos;
    }

    /**
     * MQTT客户端
     */
    private static MqttClient client;

    /**
     * 在bean容器创建之后执行该方法
     * 初始化
     */
    @PostConstruct
    public void init() {
        //连接MQTT
        this.connectMQTT();
    }
    /**
     *  连接MQTT
     */
    public MqttConnectOptions mqttConnectOptions(){
        MqttConnectOptions options = new MqttConnectOptions();;
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        //设置连接超时
        options.setConnectionTimeout(1);
        //设置心跳时间 单位为秒,向客户端发送心跳判断客户端是否在线
        options.setKeepAliveInterval(1);
        // mqttClient.reconnect(); 方法会判断这个参数
        options.setAutomaticReconnect(true);
        // 设置消息保留
        try {
            options.setWill(topic,"offline".getBytes("UTF-8"),0,true);
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        return options;
    }
    /**
     * MQTT连接客户端
     */
    public void connectMQTT() {
        try {
            log.info("MQTT连接中...");
             client = new MqttClient(broker, clientid, new MemoryPersistence());
            //回调
//            client.setCallback(new MqttCallback() {
//
//                public void connectionLost(Throwable cause) {
//                    log.info("连接断开，正在重新连接...:{}", cause.getMessage());
//                    while(true) {
//                        try {
//                            Thread.sleep(30000);
//                            // 重新连接
//                client.connect(options);
//                // 重新订阅
//                client.subscribe(topic, qos);
//                            break;
//                        } catch (Exception e) {
//                            e.printStackTrace();
//                            continue;
//                        }
//                    }
//                    log.info("重新连接成功");
//                }
//
//                public void messageArrived(String topic, MqttMessage message) {
//                    log.info("收到消息主题:{},消息内容:{},消息的推送级别次数:{}"
//                            ,topic,new String(message.getPayload()),message.getQos());
//
//                }
//
//                public void deliveryComplete(IMqttDeliveryToken token) {
//                    log.info("消息发送后的处理");
//                }
//
//            });
            client.setCallback(new MqttMessageCallback());//回调
            client.connect(mqttConnectOptions());//连接
            log.info("-----MQTT连接成功------");
            this.subscribeMQTT();//订阅
//            this.publishMQTT();//发布消息
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }
    /**
     * 订阅 MQTT 主题
     */
    public void subscribeMQTT() {
        try {
            client.subscribe(topic, qos);//订阅
            log.info("-----MQTT订阅成功------");
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 重新连接 MQTT 服务器
     */
    public void startReconnect() {
        while (true) {
            try {
                Thread.sleep(20000);
                // 重新连接
                client.connect(mqttConnectOptions());
                // 重新订阅
                client.subscribe(topic, qos);
                break;
            } catch (Exception e) {
                e.printStackTrace();
                continue;
            }
        }
    }
    /**
     * 发布 MQTT 消息
     */
    public void publishMQTT(String content) {
//        String content = "Hello MQTT";
        log.info("-----MQTT发布消息------");
        try {
            MqttMessage message = new MqttMessage();
            message.setPayload(content.getBytes());
            message.setQos(qos);
            // 连接
//            client.connect(mqttConnectOptions());
            //发布消息
            client.publish(topic, message);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }
}